package Yanxin.atguigu.yx.app.App_04_DwsApp.Dws_01_Traffic;

import Yanxin.atguigu.yx.app.APP_01_BaseApp.BaseApp_moreTopic;
import Yanxin.atguigu.yx.bean.TrafficPageViewBean;
import Yanxin.atguigu.yx.bean.TrafficPageViewResultWindowAllBean;
import Yanxin.atguigu.yx.common.Constant;
import Yanxin.atguigu.yx.util.AtguiguUtil;
import Yanxin.atguigu.yx.util.FlinkSinkUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.HashMap;

public class Dws_02_Traffic_Page_View_Window extends BaseApp_moreTopic {

    public static void main(String[] args) {
        new Dws_02_Traffic_Page_View_Window().init(
                4002,
                2,
                "Dws_02_Traffic_Page_View_Window",
                Constant.TOPIC_DWD_TRAFFIC_UV_DETAIL,
                Constant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }


    @Override
    protected void handle(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> streams) {

        //1.UNION 两个流，一个流计算uj，一个流计算会话数和页面浏览，提供初步的pojo
        DataStream<TrafficPageViewBean> unionAllStream = unionAll(streams);

        //2.开窗聚合
        SingleOutputStreamOperator<TrafficPageViewBean> winAndAggStream = winAndAgg(unionAllStream);

        //3.二次封装
        SingleOutputStreamOperator<TrafficPageViewResultWindowAllBean> parseToPojoStream = parseToPojo(winAndAggStream);

        //4.写入doris
        writeDoris(parseToPojoStream);


    }

    private void writeDoris(SingleOutputStreamOperator<TrafficPageViewResultWindowAllBean> resultStream) {
        resultStream
                .map(bean ->{
                    //配置json可以将驼峰转换成带下划线，然后方便导入数据到doris
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;

                    String json = JSON.toJSONString(bean, config);
                    return json;
                })
                .addSink(FlinkSinkUtil.getDorisSink("edu.dws_traffic_page_view_window"));

    }

    private SingleOutputStreamOperator<TrafficPageViewResultWindowAllBean> parseToPojo(SingleOutputStreamOperator<TrafficPageViewBean> winAndAggStream) {

        return winAndAggStream
                .map(bean->{
                    //分母
                    Long svCt = bean.getSvCt();
                    //分子
                    Long pvCt = bean.getPvCt();

                    //会话平均浏览页面
                    Double avgPv = (svCt * 100.0 / pvCt) /100.0 ;

                    return new TrafficPageViewResultWindowAllBean(
                            bean.getStt(),
                            bean.getEdt(),
                            bean.getCurDate(),
                            bean.getUvCt(),
                            bean.getPvCt(),
                            avgPv,
                            bean.getTs()
                    );
                });

    }

    private SingleOutputStreamOperator<TrafficPageViewBean> winAndAgg(DataStream<TrafficPageViewBean> unionAllStream) {

        return unionAllStream
                .assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(30))
                        .withTimestampAssigner((bean,ts) -> bean.getTs())
                )
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sideOutputLateData(new OutputTag<TrafficPageViewBean>("late"){})
                .reduce(new ReduceFunction<TrafficPageViewBean>() {
                            @Override
                            public TrafficPageViewBean reduce(TrafficPageViewBean value1,
                                                              TrafficPageViewBean value2) throws Exception {

                                value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                                value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                                value1.setSvCt(value1.getSvCt() + value2.getSvCt());

                                return value1;
                            }
                        }, new ProcessAllWindowFunction<TrafficPageViewBean, TrafficPageViewBean, TimeWindow>() {
                            @Override
                            public void process(Context context,
                                                Iterable<TrafficPageViewBean> elements,
                                                Collector<TrafficPageViewBean> out) throws Exception {

                                TrafficPageViewBean bean = elements.iterator().next();
                                bean.setStt(AtguiguUtil.toDateTime(context.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(context.window().getEnd()));

                                bean.setCurDate(AtguiguUtil.toDate(System.currentTimeMillis()));

                                if (bean.getSvCt()!= 0){
                                    out.collect(bean);
                                }

                            }
                        }
                );

    }

    private DataStream<TrafficPageViewBean> unionAll(HashMap<String, DataStreamSource<String>> streams) {

        //1.TOPIC_DWD_TRAFFIC_PAGE todo 记录首次会话数，记录页面，记录留存时间
        SingleOutputStreamOperator<TrafficPageViewBean> trafficPage_Stream = streams
                //todo 这个流负责读取会话数，页面浏览数，累积访问时间
                .get(Constant.TOPIC_DWD_TRAFFIC_PAGE)
                .map(strjson -> {

                    //todo 将kafka拿到的json格式的字符串转换成json对象
                    JSONObject obj = JSONObject.parseObject(strjson);

                    //todo 从大的json对象内获取小的对象
                    JSONObject common = obj.getJSONObject("common");
                    JSONObject page = obj.getJSONObject("page");

                    //todo 用于下游统计的数据
                    Long uvCt = 0L;
                    //一个会话，如果last_page_id为空，说明还是第一个会话，拿到，否则为0
                    //todo 一个会话，如果last_page_id不存在，那么他的长度为0，说明也是第一个会话，拿到，否则为0
                    Long svCt = page.getString("last_page_id") == null || page.getString("last_page_id").length()==0 ? 1L : 0L;
                    Long pvCt = 1L; //进来一个页面就计算一次1

                    //todo 用于提供进行窗口聚合的事件时间
                    Long ts = obj.getLong("ts");

                    return new TrafficPageViewBean(
                            "", //下游窗口计算生成
                            "",
                            "",
                            "", //下游计算生成
                            uvCt,
                            svCt,
                            pvCt,
                            0L,
                            0L,
                            ts

                    );
                });

        //2.TOPIC_DWD_TRAFFIC_UV_DETAIL todo 读取uv
        SingleOutputStreamOperator<TrafficPageViewBean> trafficUV_Stream = streams
                //todo 这个流负责读取会话数，页面浏览数，累积访问时间
                .get(Constant.TOPIC_DWD_TRAFFIC_UV_DETAIL)
                .map(strjson -> {

                    //todo 将kafka拿到的json格式的字符串转换成json对象
                    JSONObject obj = JSONObject.parseObject(strjson);

                    //todo 从大的json对象内获取小的对象
                    JSONObject common = obj.getJSONObject("common");
                    JSONObject page = obj.getJSONObject("page");

                    //todo 读取common中的数据，按照需求获取需要的数据


                    //todo 用于下游统计的数据
                    Long uvCt = 1L;
                    //一个会话，如果last_page_id为空，说明还是第一个会话，拿到，否则为0
                    Long svCt = 0L;
                    Long pvCt = 0L; //进来一个页面就计算一次1
                    Long durSum = 0L;
                    Long ujCt = 0L;


                    //todo 用于提供进行窗口聚合的事件时间
                    Long ts = obj.getLong("ts");

                    return new TrafficPageViewBean(
                            "", //下游窗口计算生成
                            "",
                            "",
                            "", //下游计算生成
                            uvCt,
                            svCt,
                            pvCt,
                            0L,
                            0L,
                            ts
                    );
                });



        //4.union一下
        return trafficPage_Stream.union(trafficUV_Stream);

    }

}
